with tab as (
 select billcode,
       scansitecode,
       split(concat(scantime,'\001',
       coalesce(inputtime,''),'\001',
       coalesce(subbillcode,''),'\001',
       coalesce(packagecode,''),'\001',
       coalesce(scantype,''),'\001',
       coalesce(scanuser,''),'\001',
       coalesce(operatedate,''),'\001',
       coalesce(pcs,''),'\001',
       coalesce(weight,''),'\001',
       coalesce(expresstype,''),'\001',
       coalesce(returnflag,''),'\001',
       coalesce(nextstationcode,''),'\001',
       coalesce(source,''),'\001',
       coalesce(transfercode,'')
       ),'\001') as wid_columns
from jms_ods.tab_barscan_arrival      --21天内有效
where  dt BETWEEN date_add('{{ execution_date | cst_ds }}', -20) AND '{{ execution_date | cst_ds }}'
)
insert overwrite table jms_dwd.dwd_barscan_arrival_dt
select
billcode                                                 as billcode,
scansitecode                                             as scansitecode_later,
max(wid_columns)[0]                                      as scantime_later,
max(wid_columns)[1]                                      as inputtime_later,
max(wid_columns)[2]                                      as subbillcode_later,
max(wid_columns)[3]                                      as packagecode_later,
max(wid_columns)[4]                                      as scantype_later,
max(wid_columns)[5]                                      as scanuser_later,
max(wid_columns)[6]                                      as operatedate_later,
max(wid_columns)[7]                                      as pcs_later,
cast(max(wid_columns)[8] AS DECIMAL(10, 2))              as weight_later,
max(wid_columns)[9]                                      as expresstype_later,
max(wid_columns)[10]                                     as returnflag_later,
max(wid_columns)[11]                                     as nextstationcode_later,
max(wid_columns)[12]                                     as source_later,
max(wid_columns)[13]                                     as transfercode_later,

scansitecode                                             as scansitecode_early,
min(wid_columns)[0]                                      as scantime_early,
min(wid_columns)[1]                                      as inputtime_early,
min(wid_columns)[2]                                      as subbillcode_early,
min(wid_columns)[3]                                      as packagecode_early,
min(wid_columns)[4]                                      as scantype_early,
min(wid_columns)[5]                                      as scanuser_early,
min(wid_columns)[6]                                      as operatedate_early,
min(wid_columns)[7]                                      as pcs_early,
cast(min(wid_columns)[8] AS DECIMAL(10, 2))              as weight_early,
min(wid_columns)[9]                                      as expresstype_early,
min(wid_columns)[10]                                     as returnflag_early,
min(wid_columns)[11]                                     as nextstationcode_early,
min(wid_columns)[12]                                     as source_early,
min(wid_columns)[13]                                     as transfercode_early,
'bill_scansite'                                          as dimtype,
date(max(wid_columns)[0])                                as dt
from tab
 group by billcode,scansitecode
having
    date(max(wid_columns)[0]) between date_add('{{ execution_date | cst_ds }}', -14)  AND '{{ execution_date | cst_ds }}'
DISTRIBUTE BY dt, PMOD(HASH(RAND()), 60);
